package org.example.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author zmg
 */
//@Component
public class ConsumerListener {

    @KafkaListener(topics= "lottery_invoice",groupId = "lottery")
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {

        Optional<?> message = Optional.ofNullable(record.value());

        if (message.isPresent()) {

            System.out.println("消息消费完成" + message.get());
            ack.acknowledge();
        } else {
            System.out.println("消费失败，消息重试。所有到环节，发货、更新库，都需要保证幂等。");
        }

    }




}
